package com.hyl.msg;

import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;

import org.apache.rocketmq.remoting.exception.RemotingException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RockProducer implements Iproducer {
	private DefaultMQProducer _producer = null;
	private String _groupname = "cn.js.nt.hyl";
	private String _namesrvAddr = "192.168.1.114:9876";//"192.168.0.138:9876;10.1.54.122:9876";
	private static final int TIMEOUT = 4000;
	private static final Logger log = LoggerFactory.getLogger(RockProducer.class);

	public static RockProducer getInstance(String group, String serverurl) {
		if (group == null || serverurl == null)
			return null;
		return new RockProducer(group, serverurl);
	}

	public static RockProducer getInstance(String group, String serverurl, TransactionListener transactionListener) {
		if (group == null || serverurl == null)
			return null;
		return new RockProducer(group, serverurl, transactionListener);
	}
	private RockProducer(String group, String serverurl) {
		/**         
		 *  一个应用创建一个Producer，由应用来维护此对象，可以设置为全局对象或者单例<br>   		 
		* 注意：ProducerGroupName需要由应用来保证唯一<br>         
		*  ProducerGroup这个概念发送普通的消息时，作用不大，但是发送分布式事务消息时，比较关键，         
		*   因为服务器会回查这个Group下的任意一个Producer         */
		_groupname = group;
		_namesrvAddr = serverurl;
		_producer = new DefaultMQProducer(_groupname);
		_producer.setNamesrvAddr(_namesrvAddr);
		/**         
		 * * Producer对象在使用之前必须要调用start初始化，初始化一次即可<br>         
		注意：切记不可以在每次发送消息时，都调用start方法         */
	}

	private ExecutorService _executorService;
	private TransactionListener _transactionListener;

	private RockProducer(String group, String serverurl, TransactionListener transactionListener) {
		_groupname = group;
		_namesrvAddr = serverurl;
		//载入	
		TransactionMQProducer p = new TransactionMQProducer(_groupname);

		if (transactionListener == null) {
			_transactionListener = new RockTransactionListenerImpl();
		} else {
			_transactionListener = transactionListener;

			//载入处理事务
		}
		p.setTransactionListener(_transactionListener);
		p.setExecutorService(getTransactionThreadPool());
		p.setNamesrvAddr(_namesrvAddr);
		_producer = p;

	}
	private boolean sendMsg(Message msg) {
		if (_scb != null) {
			sendMsgAsyn(msg);
		} else {
			return sendMsg1(msg);
		}
		return true;
	}
/**
 * 
 * <p>Title: sendMsg1</p>
 * <p>Description: 同步方式发送消息</p>
 * @param msg
 * @return
 * @return :boolean
 * @throws
 */
	private boolean sendMsg1(Message msg) {
		//声明并初始化一个producer 
		//需要一个producer group名字作为构造方法的参数，这里为producer1 	
		//调用producer的send()方法发送消息
		//这里调用的是同步的方式，所以会有返回结果	
		//发送完消息之后，调用shutdown()方法关闭producer
		int count = 0;
		boolean status = false;
		while (count <= 3) {
			try {
				//	_producer.send(msg, mq)
				SendResult sendResult;
				//if(_producer instanceof  TransactionMQProducer)
				if (_transactionListener == null)
					sendResult = _producer.send(msg, TIMEOUT);
				else
					sendResult = _producer.sendMessageInTransaction(msg, null);
				//  System.out.printf("%s%n", sendResult);

				if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
					status = true;
					break;
				}
			} catch (InterruptedException e) {
				log.error("MQ发送消息：" + count + "次失败，原因" + e.getMessage(), e);
			} catch (RemotingException e) {
				log.error("MQ发送消息：" + count + "次失败，原因" + e.getMessage(), e);
			} catch (MQClientException e) {
				log.error("MQ发送消息：" + count + "次失败，原因" + e.getMessage(), e);
			} catch (MQBrokerException e) {
				log.error("MQ发送消息：" + count + "次失败，原因" + e.getMessage(), e);
			} finally {
				count++;
			}
		}
		return status;
	}
	//顺序方式发送消息 (原理是只用某个queue)
	//定时方式发送
	//广播模式接收
	/**设置以后采用 异步方式发送消息**/
	private SendCallback _scb;

	public void SetCallback(SendCallback scb) {
		_scb = scb;
	}

	private void sendMsgAsyn(Message msg) {
		_producer.setRetryTimesWhenSendAsyncFailed(0);
		_producer.setSendMsgTimeout(TIMEOUT);
		if (_scb == null)
			_scb = new SendCallback() {
				@Override
				public void onSuccess(SendResult sendResult) {
					log.info("传输成功");
				}

				@Override
				public void onException(Throwable e) {
					log.error("传输失败", e);
				}
			};
		int count = 3;
		while (count > 0) {
			try {
				if (_transactionListener == null)
					_producer.send(msg, _scb);
				else
					_producer.sendMessageInTransaction(msg, _scb);
				return;
			} catch (MQClientException | RemotingException | InterruptedException e) {
				count--;
				log.error("MQ发送消息异常" + e.getMessage(), e);
			}
		}
	}

	//			Message msg = new Message("TopicTest", // topic
	//						"TagA", //tag 
	//						("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)// body 
	//				);
	public void pushMsg(String topic, String tags, String keys, byte[] msg) {
		start();
		Message message = new Message(topic, tags, keys, msg);
		sendMsg(message);
		// stop();
	}

	public void pushMsgs(List<Message> msgs) {
		start();
		for (Message msg : msgs) {
			sendMsg(msg);
		}
		// stop();

	}

	private void start() {
		try {
			_producer.start();
		} catch (MQClientException e) {
			log.error("初始化RocketMQ异常,将退出程序" + e.getMessage(), e);
			// SystemResourcesClean.closeAllResources();        
			System.exit(-1);
		}
	}



	//如果需要可以设置事务的线程池[非必须属性]
	private ExecutorService getTransactionThreadPool() {
		ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS,
				new ArrayBlockingQueue<Runnable>(3), new ThreadFactory() {
					@Override
					public Thread newThread(Runnable r) {
						Thread thread = new Thread(r);
						thread.setName("client-transaction-msg-check-thread");
						return thread;
					}
				});
		return executorService;

	}

	public void stop() {
		if (_producer != null) {

			_producer.shutdown();
			_producer = null;
		}
	}

	public void close() {
		stop();
		_transactionListener = null;
		if (_executorService != null)
			_executorService.shutdown();
		_executorService = null;
	};

}
